package com.eqxiu.gateway.user

import java.net.URL
import java.text.SimpleDateFormat
import java.util
import java.util.{Locale, Map, Properties}
import java.util.concurrent.{Executors, ScheduledExecutorService, TimeUnit}

import com.alibaba.fastjson.JSONObject
import com.eqxiu.dubbo.DubboUtil
import com.eqxiu.gateway.user.GatewayLogStream.processRecord
import com.eqxiu.js.JavaScriptProvider
import com.eqxiu.logserver.ipip.City
import com.eqxiu.logserver.util.Constant
import com.google.common.io.Resources
import com.google.gson.{JsonArray, JsonObject}
import com.maple.util.common.{JsonUtil, Mysql}
import net.sf.ehcache.{Cache, CacheManager}
import net.sf.ehcache.config.{CacheConfiguration, Configuration}
import org.apache.kafka.streams.kstream.Materialized
import org.apache.kafka.streams.scala.ImplicitConversions._
import org.apache.kafka.streams.scala._
import org.apache.kafka.streams.scala.kstream._
import org.apache.kafka.streams.{KafkaStreams, StreamsConfig}

import scala.util.control.Breaks._

case class GatewayRule(api_name: String, api_value: String, api_id: String, category_id: String, category_name: String)

case class GatewayOriginalRecord(remote_addr: String, time_local: String, cookie_JSESSIONID: String, host: String, request_method: String, uri: String, args: String, request_body: String, status: String, http_user_agent: String, http_referer: String, request_time: String, upstream_addr: String, upstream_status: String, upstream_response_time: String, body_bytes_sent: String, http_x_forwarded_for: String, upstream_http_set_cookie: String , tracker_user_id:String )

object GatewayLogStream {

    //不参与统计的接口
    var blackInterfaceArray:Array[String] = null

    implicit class SuperString(str:String){
        def clear():String = {
            if(str.trim.equals("-")){
                return ""
            }
            return str
        }

        def replaceDoubleQuotes():String =  str.replace("\"", "")

        def getClientPlatform = {
            val lowerUA = str.toLowerCase()
            var c_p = "other"
            if (lowerUA.contains("android")) {
                c_p = "Android"
            }
            else if (lowerUA.contains("ios") || lowerUA.contains("iphone")){
                c_p = "iOS"
            }
            else{
                c_p = "PC"
            }
            c_p
        }

        //规范化后的平台
        def getPlatform(): String = {
            var c_p = "PC"
            val lowerUA = str.toLowerCase

            if (str.contains("EQXiuApplet")){
                c_p = "Applet"
            }
            else if (lowerUA.contains("android")) {
                if (lowerUA.contains("yiqixiu")) {
                    c_p = "App"
                } else {
                    c_p = "wap"
                }
            } else if (lowerUA.toLowerCase.contains("iphone") || lowerUA.contains("ipad") || lowerUA.contains("ipod")) {
                if (lowerUA.contains("yiqixiu")) {
                    c_p = "App"
                } else {
                    c_p = "wap"
                }
            } else if (lowerUA.contains("windows") || lowerUA.contains("macintosh") || lowerUA.contains("windows phone") || lowerUA.contains("eqxiu-internal-mall")) {
                c_p = "PC"
            }
            return c_p
        }

        def getProductCode():String ={
            try{
                if (str.contains("EQXiuApplet")){
                    return str.split(" ")(0).split("-")(1)
                }
            }catch {
                case e: Exception =>
                    println("解析product code出现异常")
            }
            return ""
        }

        //规范化后的端
        def getTerminal()={
            val lowerUA = str.toLowerCase()
            var teriminal = "pc"
            if (lowerUA.contains("ios") || lowerUA.contains("iphone") || lowerUA.contains("ipad") || lowerUA.contains("ipod") ){
                teriminal = "ios"
            }
            else if (lowerUA.contains("android") ){
                teriminal = "android"
            }
            teriminal
        }
    }

    import Serdes._

    val sdf = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z", Locale.US);
    val ChinaSDfFull = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    val ChinaSDfDay = new SimpleDateFormat("yyyy-MM-dd");
    val ChinaSDfMonth = new SimpleDateFormat("yyyy-MM");

    private val scheduledExecutorService = Executors.newSingleThreadScheduledExecutor
    var gatewayRules: Array[GatewayRule] = null
    var city: City = null;

    def initEventConf(mysqlUrl: String, mysqlUser: String, mysqlPassword: String) {
        val mysql = new Mysql(mysqlUrl, mysqlUser, mysqlPassword)
        val gatewayRules0 = mysql.queryForList("SELECT a.category_id, b.category_name, a.api_id, c.api_name, c.api_value FROM gateway_category_api a JOIN gateway_category b ON a.category_id = b.id JOIN gateway_api c ON a.api_id = c.id").toArray.map { row =>
            val m = row.asInstanceOf[java.util.Map[String, Object]]
            val api_name = m.get("api_name").toString
            val api_value = m.get("api_value").toString
            val api_id = m.get("api_id").toString
            val category_id = m.get("category_id").toString
            val category_name = m.get("category_name").toString
            GatewayRule(api_name, api_value, api_id, category_id, category_name)
        }
        gatewayRules = gatewayRules0
    }

    private def parseTime(time_local: String) = {

        var dt = new java.util.Date()
        if ("-" != time_local) {
            // 解析时间
            try {
                dt = sdf.parse(time_local)
            }
            catch {
                case e: Exception =>
                    println(e)
            }
        }
        val fdtFull = ChinaSDfFull.format(dt)
        val fdtDay = ChinaSDfDay.format(dt)
        val fdtMonth = ChinaSDfMonth.format(dt)
        val dtTuple = (fdtFull, fdtDay, fdtMonth,dt.getTime)

        dtTuple
    }

    // 加载IP库
    def initIpDatabase(ipFilePath: String) {
        try {
            city = new City(ipFilePath)
        } catch {
            case e: Exception =>
                System.err.println("未找到IP库文件：" + ipFilePath)
        }
    }

    //解析网关日志
    def processRecord(line: String): JsonObject = {
        val json = new JsonObject()
        val values = line.split("#\\|#")

        if (values.length < 19) {
            return null
        }

        val originalRecord = GatewayOriginalRecord(values(0), values(1), values(2), values(3), values(4), values(5), values(6), values(7), values(8), values(9), values(10), values(11), values(12), values(13), values(14), values(15), values(16), values(17),values(18))

        val url = originalRecord.host + originalRecord.uri

        if( originalRecord.uri == null || blackInterfaceArray.contains(originalRecord.uri)){
//            println("debug WARNING: black interface="+originalRecord.uri)
            return null;
        }

        var gatewayRule: GatewayRule = null
        breakable(
            gatewayRules.foreach { gatewayRuleTmp =>
                if (originalRecord.uri.matches(gatewayRuleTmp.api_value)) {
                    gatewayRule = gatewayRuleTmp
                    break()
                }
            })

//      经和郭老板沟通后，此处可以过滤掉未在max后台登记的接口
        if (gatewayRule == null) {
            //不再往下游的topic发送日志
            //      println("no rules , continue")
            return null
        }

        var session_id = GatewayLogUtil.getSessionId(originalRecord)

        json.addProperty("s_i", session_id)

        json.addProperty("ip", originalRecord.remote_addr)
        json.addProperty("cookie_user_id", originalRecord.tracker_user_id)

        //用户id
        try{
            json.addProperty("user_id", GatewayLogUtil.getUserId(session_id,originalRecord))
            // println("debug WARNING: invoke count: tracker_invoke_count=" +  GatewayLogUtil.tracker_invoke_count +"  dubbo_invoke_count="+GatewayLogUtil.dubbo_invoke_count)
        }catch{
            case e: Throwable =>
                println("debug WARNING: user_id 获取异常,此时json数据为:" +  json.toString)
                e.printStackTrace()
        }

        if( gatewayRule == null ){
            //接口id
            json.addProperty("api_id", "")
            //接口标识
            json.addProperty("api_value", "")
            //接口名称
            json.addProperty("api_name","")
            //接口分类
            json.addProperty("api_category", "0")
        }else {
            //接口id
            json.addProperty("api_id", gatewayRule.api_id)
            //接口标识
            json.addProperty("api_value", gatewayRule.api_value)
            //接口名称
            json.addProperty("api_name", gatewayRule.api_name)
            //接口分类
            json.addProperty("api_category", gatewayRule.category_id)
        }

        //    接口分类id 不需要
        //    json.addProperty("category_id",gatewayRule.category_id)
        //    接口分类名称 不需要
        //    json.addProperty("category_name",gatewayRule.category_name)
        json.addProperty("host", originalRecord.host)
        json.addProperty("uri", originalRecord.uri)

        json.addProperty("url", url)
        json.addProperty("ref", originalRecord.http_referer)

        val ua = originalRecord.http_user_agent

        json.addProperty("u_a", ua)

        val c_p = ua.getClientPlatform

        json.addProperty("c_p", c_p)

        val platform = ua.getPlatform()

        json.addProperty("platform", platform)

        val productCode = ua.getProductCode()

        json.addProperty("product_code", productCode)

        val terminal = ua.getTerminal()
        json.addProperty("terminal", terminal)

        var g_params = ""
        if( originalRecord.args != null )
        {
            g_params = originalRecord.args.clear().replaceDoubleQuotes()
        }
        if(g_params.length > 1024){
            g_params = "the param is too long"
        }
        json.addProperty("g_params",g_params)

        var p_params = ""
        if ( originalRecord.request_body != null)
        {
            p_params = originalRecord.request_body.clear().replaceDoubleQuotes()
        }
        if(p_params.length > 1024){
            p_params = "the param is too long"
        }
        json.addProperty("p_params",p_params)


        var model = ""
        var os = ""

        try {
            val j = JavaScriptProvider.getDeviceInfo(originalRecord.http_user_agent)
            model = j.getJSONObject("device").getString("model")
            os = j.getJSONObject("os").getString("name")
        } catch {
            case e: Throwable =>
                println("error:" + originalRecord.http_user_agent)
                e.printStackTrace()
        }

        //设备型号
        json.addProperty("mod", model)
        //平台
        json.addProperty("os", os)
        //    json.addProperty("ua",originalRecord.http_user_agent)
        json.addProperty("ip", originalRecord.remote_addr)
        try {
            val info = city.find(originalRecord.remote_addr)

            json.addProperty("cou", info(0))
            if( "中国".equals(info(1)) ){
                json.addProperty("pro", "" )
            }else{
                json.addProperty("pro", info(1))
            }

            json.addProperty("cit", info(2))
            var m_o = ""
            if (info != null && info.length >= 5) {
                m_o = info(4)
            }
            json.addProperty("m_o", m_o) // mobile operator
            json.addProperty("m_o_2", Constant.mobileOperatorMap.getOrDefault(m_o, "Others"))

        } catch {
            case e: Throwable =>
                println(e)
        }
        //运营商
        json.addProperty("request_time", originalRecord.request_time)
        json.addProperty("status", originalRecord.status)

        val datetimeTuple = parseTime(originalRecord.time_local)
        json.addProperty("datetime", datetimeTuple._1)
        json.addProperty("day", datetimeTuple._2)
        json.addProperty("month", datetimeTuple._3)
        json.addProperty("s_t", datetimeTuple._4)

        return json
    }

    def main(args: Array[String]): Unit = {

        val confPath = args(0)
        com.maple.util.common.CB.init(confPath)

        val applicationID = com.maple.util.common.CB.s("application_id")
        val bootstrapServers = com.maple.util.common.CB.s("kafka_bootstrap_servers")
        val inputTopic = com.maple.util.common.CB.s("input_topic")
        val outputTopic = com.maple.util.common.CB.s("output_topic")
        val dubboConsumerPath = com.maple.util.common.CB.s("user_dubbo_file_path")

        val mysqlUrl = com.maple.util.common.CB.mysql.url
        val mysqlUsername = com.maple.util.common.CB.mysql.username
        val mysqlPassword = com.maple.util.common.CB.mysql.password
        val ipFilePath = com.maple.util.common.CB.s("ipFilePath")
        val jsFilePath = com.maple.util.common.CB.s("jsFilePath")

        blackInterfaceArray = com.maple.util.common.CB.s("black_interface_list").split(",")

        DubboUtil.init(dubboConsumerPath)
        initEventConf(mysqlUrl, mysqlUsername, mysqlPassword)
        initIpDatabase(ipFilePath)
        Constant.init(ipFilePath)
        JavaScriptProvider.initJavaScript(jsFilePath)
        GatewayLogUtil.initCache()

        scheduledExecutorService.scheduleAtFixedRate(new Runnable {
            override def run(): Unit = {
                initEventConf(mysqlUrl, mysqlUsername, mysqlPassword)
                println("sync database configuration")
                gatewayRules.foreach(println)
            }
        }, 0, 10, TimeUnit.MINUTES
        )

        val props: Properties = {
            val p = new Properties()
            p.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationID)
            p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers)
            p.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, new DefaultEventTimeExtractor().getClass)
            p
        }

        val builder: StreamsBuilder = new StreamsBuilder
        val textLines: KStream[String, String] = builder.stream[String, String](inputTopic)
        val wordCounts = textLines.flatMap { (key, value) =>
            val result = processRecord(value)
            if (result == null) {
                None
            } else {
                val value1 = processRecord(value).toString
                Some((key, value1))
            }
        }

        wordCounts.to(outputTopic)

        val streams: KafkaStreams = new KafkaStreams(builder.build(), props)
        streams.start()

        sys.ShutdownHookThread {
            streams.close(10, TimeUnit.SECONDS)
        }
    }

}